package com.yanggu.flink.tableapi_sql.define_table

import com.yanggu.flink.datastream_api.pojo.SensorReading
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._

//使用DataStream的方式定义Table
//详情见官方文档https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/table/data_stream_api/#handling-of-insert-only-streams
object DataStreamForTable {

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment

    val settings = EnvironmentSettings
      .newInstance()
      .inStreamingMode()
      .build()

    val tableEnvironment = StreamTableEnvironment.create(environment, settings)

    val dataStream = environment
      .readTextFile(getClass.getResource("/sensor.txt").getPath)
      .map(data => {
        val strings = data.split(",")
        SensorReading(strings(0), strings(1).toLong, strings(2).toDouble)
      })

    //使用这种方式定义table, 样例类中的字段名称、数据类型和顺序就是表与之对应的表结构
    //(
    //  `id` STRING,
    //  `timestamp` BIGINT NOT NULL,
    //  `temperature` DOUBLE NOT NULL
    //)
    tableEnvironment.fromDataStream(dataStream).printSchema()

    //需要导入隐式转换import org.apache.flink.table.api._
    //基于名称
    //使用表达式通过样例类的字段名来定义表的字段
    //这个已经过期了, 不推荐使用
    //tableEnvironment.fromDataStream(dataStream, 'id, 'timestamp as 'ts)
    //(
    //  `id` STRING,
    //  `ts` BIGINT NOT NULL,
    //  `temp` DOUBLE NOT NULL
    //)
    tableEnvironment.fromDataStream(dataStream).as("id", "ts", "temp").printSchema()

    //基于位置的对应, 常见于元祖
    //对于元祖, 表中的字段名就是_1、_2
    //(
    //  `id` BIGINT,
    //  `name` STRING
    //)
    tableEnvironment.fromDataStream(environment.fromElements((1L, "张三"))).as("id", "name").printSchema()

    //对于原子数据类型就是字段名f0
    //貌似不支持这种写法
    //val singleTable = tableEnvironment.fromDataStream(singleDataStream, 'f0 as("f0", "name"))
    //(
    //  `name` STRING
    //)
    tableEnvironment.fromDataStream(environment.fromElements("Alice")).as("name").printSchema()

    //使用Schema的方式来定义
    val schema = Schema
      .newBuilder()
      .column("id", DataTypes.STRING())
      .column("timestamp", DataTypes.BIGINT())
      .column("temperature", DataTypes.DOUBLE())
      .build()
    tableEnvironment.fromDataStream(dataStream, schema)

    //在TableEnv中注册表
    tableEnvironment.createTemporaryView("sensor", dataStream, schema)

    tableEnvironment.sqlQuery(
      """
        |SELECT
        |   id,
        |   MAX(`timestamp`),
        |   MAX(temperature)
        |FROM
        |   sensor
        |GROUP BY
        |   id
        |""".stripMargin)
      .execute()
      .print()

  }

}
